package spark.network;

import spark.utility.IncrementalLongIdFactory;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ClockSynchronizer {
	private final ReadWriteLock lock = new ReentrantReadWriteLock();
	private final Lock readLock = lock.readLock();
	private final Lock writeLock = lock.writeLock();

	private HashMap<Long, HashMap<Long, SyncData>> syncDataMap = new HashMap<Long, HashMap<Long, SyncData>>();
	private IncrementalLongIdFactory requestNumberFactor = new IncrementalLongIdFactory();

	public long makeSyncData(long session) {
		SyncData syncData = new SyncData();
		syncData.requestNumber = requestNumberFactor.getNextId();
		syncData.serverSendTime = System.currentTimeMillis();
		writeLock.lock();
		try {
			if(!syncDataMap.containsKey(session)) {
				syncDataMap.put(session, new HashMap<Long, SyncData>());
			}
			syncDataMap.get(session).put(syncData.requestNumber, syncData);
		}
		finally {
			writeLock.unlock();
		}
		return syncData.requestNumber;
	}

	public void updateSyncData(long session, long requestNumber, long serverReceiveTime, long clientLocalTime) {
		readLock.lock();
		SyncData syncData = null;
		try {
			syncData = syncDataMap.get(session).get(requestNumber);
		}
		finally {
			readLock.unlock();
		}
		syncData.serverReceiveTime = serverReceiveTime;
		syncData.clientLocalTime = clientLocalTime;
	}

	public int getCompleteSyncDataCount(long session) {
		int completeCount = 0;
		readLock.lock();
		try {
			for(SyncData syncData : syncDataMap.get(session).values()) {
				if(syncData.isComplete()) {
					completeCount++;
				}
			}
			return completeCount;
		}
		finally {
			readLock.unlock();
		}
	}

	public SyncData getShortestSyncData(long session) {
		SyncData shortest = null;
		readLock.lock();
		try {
			for(SyncData syncData : syncDataMap.get(session).values()) {
				if(syncData.isComplete()) {
					if(shortest == null ||
					   syncData.getRoundTripTime() < shortest.getRoundTripTime()) {
						shortest = syncData;
					}
				}
			}
		}
		finally {
			readLock.unlock();
		}
		return shortest;
	}

	public void clearSyncData(long session) {
		writeLock.lock();
		try {
			syncDataMap.remove(session);
		}
		finally {
			writeLock.unlock();
		}
	}

	public static class SyncData {
		public Long requestNumber = null;
		public Long serverSendTime = null;
		public Long serverReceiveTime = null;
		public Long clientLocalTime = null;

		public long getRoundTripTime() {
			return serverReceiveTime - serverSendTime;
		}

		public long getAdjustedClientLocalTime() {
			return clientLocalTime + getRoundTripTime() / 2;
		}

		public boolean isComplete() {
			return clientLocalTime != null;
		}

		@Override
		public boolean equals(Object o) {
			if(this == o) return true;
			if(o == null || getClass() != o.getClass()) return false;

			SyncData syncData = (SyncData) o;

			if(requestNumber != syncData.requestNumber) return false;

			return true;
		}

		@Override
		public int hashCode() {
			return (int) (requestNumber ^ (requestNumber >>> 32));
		}
	}
}
